package xdb;

import com.goldhuman.Common.Marshal.MarshalException;
import com.goldhuman.Common.Marshal.OctetsStream;
import mysql.MySqlMgr;
import xdb.TRecord.State;

import java.io.ByteArrayInputStream;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;

final class CheckpointCacheInfo {

    private final String tableName;
    private final ConcurrentMap<String, Treble<State, OctetsStream, Long>> record = xdb.util.Misc.newConcurrentMap();
    private long hitCount = 0;

    CheckpointCacheInfo(String tableName) {
        this.tableName = tableName;
    }

    public String getTableName() {
        return tableName;
    }

    public void addRecord(String key, State state, OctetsStream oct, long flushTime) {
        if (xdb.Trace.isDebugEnabled()) {
            xdb.Trace.debug(tableName + " record:" + record.size() + " key:" + key);
        }
        record.put(key, new Treble<State, OctetsStream, Long>(state, oct, flushTime));
    }

    public boolean existInCache(String key) {
        if (record.containsKey(key) == false) {
            return false;
        }
        switch (record.get(key).first) {
            case INDB_REMOVE:
            case REMOVE:
            case INDB_GET:
            case INDB_ADD:
            case ADD:
                return true;
            default:
                return false;
        }
    }

    public OctetsStream find(String key) {
        if (record.containsKey(key) == false) {
            return null;
        }
        switch (record.get(key).first) {
            case INDB_REMOVE:
            case REMOVE:
                return null;
            case INDB_GET:
            case INDB_ADD:
            case ADD:
                hitCount++;
                if (hitCount % 100 == 0) {
                    xdb.Trace.info(">>> " + tableName + " >>> " + Long.parseLong(key, 16) + " >>> hitCount " + hitCount);
                }
                return record.get(key).second;
            default:
                return null;
        }
    }

    public int cacheCount() {
        return record.size();
    }

    public boolean isZero(String key) {
        for (int i = 0; i < key.length(); i++) {
            if (key.charAt(i) != '0') {
                return false;
            }
        }
        return true;
    }

    public void save(int maxCount) {
        if (record.size() == 0) {
            return;
        }
        int initRecordCount = record.size();
        HashSet<String> failedRecordKey = new HashSet<String>();
        Iterator<Map.Entry<String, Treble<State, OctetsStream, Long>>> iter = record.entrySet().iterator();

        int count = 0;
        if (initRecordCount >= CheckpointCache.PROCESS_BATCH || maxCount == Integer.MAX_VALUE) {
            long now = System.currentTimeMillis();
            Connection conn = null;
            try {
                conn = MySqlMgr.GetInstance().GetWriteConn();
                PreparedStatement pstReplace = conn.prepareStatement("REPLACE INTO " + tableName + " VALUES(?, ?)");
                pstReplace.setQueryTimeout(10);//批量更新超时设置成10秒，避免线程卡主
                PreparedStatement pstRemove = conn.prepareStatement("DELETE FROM " + tableName + " WHERE k=?");
                pstRemove.setQueryTimeout(10);//批量更新超时设置成10秒，避免线程卡主
                while (iter.hasNext()) {
                    Map.Entry<String, Treble<State, OctetsStream, Long>> entry = iter.next();
                    if (entry.getValue().third >= now && maxCount < Integer.MAX_VALUE) {
                        continue;
                    }
                    if (isZero(entry.getKey())) {
                        Trace.error("Key is Zero! Key=" + entry.getKey() + " tableName=" + tableName);
                        failedRecordKey.add(entry.getKey());
                        continue;
                    }
                    if (entry.getValue().first == State.INDB_GET || entry.getValue().first == State.INDB_ADD || entry.getValue().first == State.ADD) {
                        //先放入失败列表，成功提交再清理
                        failedRecordKey.add(entry.getKey());
                        pstReplace.setString(1, entry.getKey());
                        if (Xdb.getInstance().isJson()) {
                            pstReplace.setString(2, entry.getValue().second.unmarshalJson());
                        } else {
                            pstReplace.setBinaryStream(2, new ByteArrayInputStream(entry.getValue().second.array()), entry.getValue().second.size());
                        }
                        pstReplace.addBatch();
                    } else if (entry.getValue().first == State.INDB_REMOVE) {
                        failedRecordKey.add(entry.getKey());
                        pstRemove.setString(1, entry.getKey());
                        pstRemove.addBatch();
                    } else {
                        continue;
                    }

                    // 批量处理最多执行PROCESS_MAX_COUNT,FinalCheckPoint执行所有的数据
                    if (++count >= maxCount) {
                        break;
                    }
                }
                if (count > 0) {
                    pstReplace.executeBatch();
                    pstRemove.executeBatch();
                    conn.commit();
                }
                pstReplace.close();
                pstRemove.close();
                //在执行批量存储过程中，有新数据加入，删除就会导致数据不一致，所以只删除更新时间<now的数据
                for (String key : failedRecordKey) {
                    if (record.get(key).third >= now && maxCount < Integer.MAX_VALUE) {
                        continue;
                    }
                    record.remove(key);
                }
                failedRecordKey.clear();
                long useTime = System.currentTimeMillis() - now;
                if (useTime > MySqlMgr.MS_DEBUG) {
                    if (Trace.isDebugEnabled()) {
                        Trace.debug("Execute Batch Use " + useTime + " tableName is " + tableName);
                    }
                }
                if (useTime > MySqlMgr.MS_INFO * 2) {
                    Trace.info("Execute Batch Use " + useTime + " tableName is " + tableName);
                }
            } catch (Exception e) {
                if (conn != null) {
                    try {
                        conn.rollback();
                        xdb.Trace.error("save exec Batch Rollback Success " + tableName);
                    } catch (Exception e1) {
                        xdb.Trace.error("save exec Batch Rollback Exception " + tableName, e1);
                    }
                }
                xdb.Trace.error("save exec Batch ReplaceSQL Exception " + tableName, e);
            } finally {
                if (conn != null) {
                    try {
                        conn.close();
                    } catch (SQLException e) {
                        xdb.Trace.error("write conn close", e);
                    }
                }
            }
        }

        // 数量小于等于PROCESS_MAX_COUNT或者批量执行失败，才进行单条执行
        if (initRecordCount < CheckpointCache.PROCESS_BATCH || failedRecordKey.size() > 0 || maxCount == Integer.MAX_VALUE) {
            int countInit = count;
            long now = System.currentTimeMillis();
            iter = record.entrySet().iterator();
            while (iter.hasNext()) {
                boolean bSuccess = true;
                Map.Entry<String, Treble<State, OctetsStream, Long>> entry = iter.next();
                if (entry.getValue().third >= now && maxCount < Integer.MAX_VALUE) {
                    continue;
                }
                if (isZero(entry.getKey())) {
                    Trace.error("Key is Zero! Key=" + entry.getKey() + " tableName=" + tableName);
                    failedRecordKey.add(entry.getKey());
                    continue;
                }
                switch (entry.getValue().first) {
                    case INDB_GET:
                    case INDB_ADD:
                    case ADD:
                        String replaceSql = "REPLACE INTO " + tableName + " VALUES('" + entry.getKey() + "', ?)";
                        if (MySqlMgr.GetInstance().ExecUpdate(replaceSql, entry.getValue().second) < 0) {
                            if (MySqlMgr.GetInstance().ExecUpdate(replaceSql, entry.getValue().second) < 0) {
                                // 两次执行失败，放入队列
                                failedRecordKey.add(entry.getKey());
                                bSuccess = false;
                            }
                        }
                        break;
                    case INDB_REMOVE:
                        String deleteSql = "DELETE FROM " + tableName + " WHERE k='" + entry.getKey() + "'";
                        if (MySqlMgr.GetInstance().ExecUpdate(deleteSql, null) < 0) {
                            if (MySqlMgr.GetInstance().ExecUpdate(deleteSql, null) < 0) {
                                // 两次执行失败，放入队列
                                failedRecordKey.add(entry.getKey());
                                bSuccess = false;
                            }
                        }
                        break;
                    default:
                }
                //执行成功并且没有更新过就删除
                if (bSuccess && (record.get(entry.getKey()).third < now || maxCount == Integer.MAX_VALUE)) {
                    iter.remove();
                }
                //每次执行最多成功执行PROCESS_SINGLE条,如果是FinalCheckPoint那么就需要处理所有的数据
                if (bSuccess && ++count - countInit >= CheckpointCache.PROCESS_SINGLE && maxCount < CheckpointCache.PROCESS_MAX_COUNT) {
                    break;
                }
            }
        }

        if (failedRecordKey.size() > 0) {
            xdb.Trace.error(tableName + " success:" + count + " failed:" + failedRecordKey.size() + " left:" + record.size());
            for (String key : failedRecordKey) {
                Treble<State, OctetsStream, Long> info = record.get(key);
                if (info != null && info.second != null) {
                    try {
                        String jsonStr = info.second.unmarshalJson();
                        if (jsonStr.length() > 1000) {
                            xdb.Trace.error(tableName + " key=" + key + " json size=" + jsonStr.length());
                        } else {
                            xdb.Trace.error(tableName + " key=" + key + " json=" + jsonStr);
                        }
                    } catch (MarshalException e) {
                        xdb.Trace.error(tableName + " key=" + key, e);
                    }
                }
                if (MySqlMgr.failedClearMap.get(tableName + key) != null) {
                    record.remove(key);
                    xdb.Trace.error(tableName + " key=" + key + " is removed!");
                }
            }
        } else {
            xdb.Trace.info(tableName + " success:" + count + " left:" + record.size());
        }
    }
}
